/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package org.apache.flume.sink;

import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.flume.sink.BranchIOManager.BranchFileItem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class BranchPathConfig {
	
	private static final Logger logger = LoggerFactory.getLogger(BranchPathConfig.class);
	
	private static final String BRANCH_PATH_CONF_PATH = "conf/branchPath.properties";
	
	
	public static boolean configUpdated = false;
	
	private static Map<String, BranchFileItem> configPathes = null;
	
	private static long confFileLastModified = 0L;
	
	private static ScheduledExecutorService scheduleService = Executors.newScheduledThreadPool(1);
	
	private static final long checkConfInterval = 30000L;
	
	private static File confFile = null;
	
	static{
		
		try {
			loadPathConf(BRANCH_PATH_CONF_PATH);
		} catch (IOException ioe) {
			logger.error("Init - Load branch path conf eror: {}", ioe);
		}
		
		scheduleService.scheduleAtFixedRate(new reloadConfFile(), 
				checkConfInterval, checkConfInterval, TimeUnit.MILLISECONDS);
	}
	
	
	public static class reloadConfFile implements Runnable{
		public void run(){
			
			confFile = new File(BRANCH_PATH_CONF_PATH);
			
			if(confFileLastModified != confFile.lastModified()){
				try {
					
					loadPathConf(BRANCH_PATH_CONF_PATH);
					
					configUpdated = true;
					
				} catch (IOException ioe) {
					logger.error("Refresh - Load branch path configuration eror: {}", ioe);
				}  
			}
			return;
		}
	}
	
	
	public static String getDirectory(String pathKey){
		return configPathes.get(pathKey).getDirectory();
	}
	
	public static void resetUpdateStatu(){
		configUpdated = false;
	}
	
	public static boolean hasAPathUpdated(String pathKey){
		if(!configPathes.containsKey(pathKey)){
			return false;
		}
		return configPathes.get(pathKey).isUpdated();
	}
	
	public static void resetAPathStatu(String pathKey){
		configPathes.get(pathKey).setUpdated(false);
	}
	
	public static boolean containsKey(String pathKey){
		return configPathes.containsKey(pathKey);
	}

	private static void loadPathConf(String confFilePath) throws IOException {
		
		confFile = new File(confFilePath);
		
		InputStream in = null;
		try {
			in = new BufferedInputStream(new FileInputStream(confFile));
			Properties loadedProperties = new Properties();
			loadedProperties.load(in);
			
			if (configPathes == null) {
				configPathes = new HashMap<String, BranchFileItem>();
				for (Object key : loadedProperties.keySet()) {
					configPathes.put(key.toString(),
							new BranchFileItem(false, loadedProperties.getProperty(key.toString())));
					logger.info("Load path configure: ADD NEW key-{} value-{}"
							, key.toString(), loadedProperties.getProperty(key.toString()));
				}
			} else {
				updatePathConfigs(loadedProperties);
				logger.info("Update path configure over!");
			}
			
			confFileLastModified = confFile.lastModified();
			
		} finally {
			if (in != null) {
				in.close();
			}
		}
	}

	private static void updatePathConfigs(Properties reloadedProperties) {
		for (Object key : reloadedProperties.keySet()) {
			String newConfDirectory = reloadedProperties.getProperty(key.toString());
			BranchFileItem item = configPathes.get(key.toString());
			if (item != null) {
				if (item.fileHasChanged(newConfDirectory)) {
					item.setDirectory(newConfDirectory);
					item.setUpdated(true);
					logger.info("Update Path config: UPDATE OLD key-{} value-{}.", key.toString(), newConfDirectory);
				}
			} else {
				configPathes.put(key.toString(), new BranchFileItem(false, newConfDirectory));
				logger.info("Update Path config: ADD NEW key-{} value-{}", key.toString(), newConfDirectory);
			}
		}

		for (Iterator<String> it = configPathes.keySet().iterator(); it.hasNext();) {
			String key = it.next();
			if (!reloadedProperties.containsKey(key)) {
				configPathes.remove(key);
				logger.info("Update Path config: DELETE OLD key-{}", key);
			}
		}
	}

}
